package pro.shushi.pamirs.demo.core.mq;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.MessageClientExt;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import pro.shushi.pamirs.demo.api.enumeration.ItemStockStatusEnum;
import pro.shushi.pamirs.demo.api.model.DemoItem;
import pro.shushi.pamirs.framework.connectors.event.annotation.NotifyListener;
import pro.shushi.pamirs.framework.connectors.event.api.NotifyEventListener;
import pro.shushi.pamirs.framework.connectors.event.engine.NotifyEvent;
import pro.shushi.pamirs.framework.connectors.event.enumeration.ConsumerType;
import pro.shushi.pamirs.meta.annotation.fun.extern.Slf4j;
import pro.shushi.pamirs.meta.api.session.PamirsSession;
import pro.shushi.pamirs.meta.common.spring.BeanDefinitionUtils;
import pro.shushi.pamirs.meta.constant.FieldConstants;
import pro.shushi.pamirs.meta.util.JsonUtils;
import pro.shushi.pamirs.meta.util.TypeUtils;

import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * WorkflowMessageTriggerListener
 *
 * @author zxf
 */
@Slf4j
@Component
@NotifyListener(topic = DemoItemMQProducer.PAMIRS_MESSAGE_DEMO_ITEM_TOPIC, tags = SubscriptionData.SUB_ALL, consumerType = ConsumerType.ORDERLY)
public class DemoItemMessageDemoItemListener implements NotifyEventListener {

    public static final int DEMO_ITEM_MSG_EXPIRE_TIME = 24;//unit:hour
    public static final String DEMO_ITEM_MSG_SUCC = "succ";
    public static final String DEMO_ITEM_MSG_RUNNING = "running";

    @Override
    public void consumer(NotifyEvent event) {

        try {
            if (null == event) {
                return;
            }
            MessageClientExt ext = (MessageClientExt) event.getExtend();
            log.info("mq-consumer-traceId:" + PamirsSession.getRequestVariables().getTraceId());
            StringRedisTemplate redisTemplate = BeanDefinitionUtils.getBean(StringRedisTemplate.class);
            String msgKey = DemoItemMQProducer.PAMIRS_MESSAGE_DEMO_ITEM_TOPIC + ext.getMsgId();
            String status = redisTemplate.opsForValue().get(msgKey);
            if (StringUtils.isNotBlank(status) && status.equals(DEMO_ITEM_MSG_SUCC)) {
                return;
            }
            if (StringUtils.isNotBlank(status) && status.equals(DEMO_ITEM_MSG_RUNNING)) {
                return;
            }
            redisTemplate.opsForValue().set(msgKey, DEMO_ITEM_MSG_RUNNING, DEMO_ITEM_MSG_EXPIRE_TIME, TimeUnit.HOURS);

            Map<String, String> properties = event.getProperties();
            String msgId = properties.getOrDefault(DemoItemMQProducer.PAMIRS_MESSAGE_DEMO_ITEM_TOPIC, null);
            if (StringUtils.isBlank(msgId)) {
                return;
            }
            Map<String, Object> dataMap = JsonUtils.parseMap((String) event.getBody());
            Long itemId = TypeUtils.createLong(dataMap.get(FieldConstants.ID));
            DemoItem item = new DemoItem().queryById(itemId);
            if(item != null){
                item.setStockStatus(ItemStockStatusEnum.ACTIVE);
                item.setInventoryQuantity(item.getInventoryQuantity()+1);
                item.updateById();
            }
            log.error("消息触发消费成功: msgId:[{}] body:[{}]", JsonUtils.toJSONString(event.getProperties()), event.getBody());
            redisTemplate.opsForValue().set(msgKey, DEMO_ITEM_MSG_SUCC, DEMO_ITEM_MSG_EXPIRE_TIME, TimeUnit.HOURS);
        } catch (Throwable ex) {
            log.error("消息触发消费失败", ex);
            throw ex;
        }
    }
}